#pragma once
#include <iostream>
#include <cstdio>
#include <cstring>
#include <string>
#include <vector>
#include <unordered_map>
#include <functional>
#include <assert.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <arpa/inet.h>
#include <netinet/in.h>
#include <ctime>
#include <unistd.h>
#include <fcntl.h>
#include <sys/epoll.h>
#include <memory>
#include <sys/eventfd.h>
#include <sys/timerfd.h>
#include <thread>
#include <mutex>
#include <condition_variable>

// 日志模块----------------------------------------------------------------------------------------
#define INF 0
#define DBG 1
#define ERR 2
#define LOG_LEVEL INF

#define LOG(level, format, ...)                                                                                        \
    do                                                                                                                 \
    {                                                                                                                  \
        if (level < LOG_LEVEL)                                                                                         \
            break;                                                                                                     \
        time_t t = time(NULL);                                                                                         \
        struct tm *ltm = localtime(&t);                                                                                \
        char tmp[32] = {0};                                                                                            \
        strftime(tmp, 31, "%H:%M:%S", ltm);                                                                            \
        fprintf(stdout, "[%p %s %s:%d] " format "\n", (void *)pthread_self(), tmp, __FILE__, __LINE__, ##__VA_ARGS__); \
    } while (0)

#define INF_LOG(format, ...) LOG(INF, format, ##__VA_ARGS__)
#define DBG_LOG(format, ...) LOG(DBG, format, ##__VA_ARGS__)
#define ERR_LOG(format, ...) LOG(ERR, format, ##__VA_ARGS__)

// 缓冲区模块----------------------------------------------------------------------------------------
class Buffer
{
private:
    uint64_t _read_idx;        // 读位置
    uint64_t _write_idx;       // 写位置
    std::vector<char> _buffer; // 缓冲区数组

public:
    // 构造和析构函数
    Buffer() : _read_idx(0), _write_idx(0) {}
    ~Buffer() = default;

    // 获取数组起始地址
    void *Address() { return &*(_buffer.begin()); }
    // 获取读位置地址
    void *ReadPosition() { return (char*)Address() + _read_idx; }
    // 获取写位置地址
    void *WritePosition() { return (char*)Address() + _write_idx; }
    // 获取头部剩余空间位置
    uint64_t HeadAbleSize() { return _read_idx; }
    // 获取尾部剩余空间位置
    uint64_t TailAbleSize() { return _buffer.size() - _write_idx; }
    // 获取可读数据大小
    uint64_t ReadAbleSize() { return _write_idx - _read_idx; }

    // 读位置偏移
    void MoveReadIdx(uint64_t len)
    {
        assert(len <= ReadAbleSize());
        _read_idx += len;
    }
    // 写位置偏移
    void MoveWriteIdx(uint64_t len)
    {
        assert(len <= TailAbleSize());
        _write_idx += len;
    }
    // 确保空闲空间足够
    void EnsuerWriteSpace(uint64_t len)
    {
        // 末尾空间足够
        if (len <= TailAbleSize())
            return;
        // 整体可用空间足够
        if (len <= HeadAbleSize() + TailAbleSize())
        {
            uint64_t rsz = ReadAbleSize();
            std::copy((char *)ReadPosition(), (char *)WritePosition(), (char *)Address());
            _read_idx = 0;
            _write_idx = rsz;
        }
        // 需要扩容
        else
        {
            _buffer.resize(_write_idx + len);
        }
    }

    // 向缓冲区写入数据
    void Write(const void *buf, uint64_t len)
    {
        if (len == 0)
            return;

        // 确保可用空间是足够的
        EnsuerWriteSpace(len);
        // 写入数据
        std::copy((char *)buf, (char *)buf + len, (char*)WritePosition());
        // 移动写位置
        MoveWriteIdx(len);
    }
    void WriteString(const std::string &str, uint64_t len)
    {
        Write(&*(str.begin()), str.size());
    }
    void WriteBuffer(Buffer &buf)
    {
        Write(buf.ReadPosition(), buf.ReadAbleSize());
    }
    // 从缓冲区读取数据
    void Read(void *buf, uint64_t len)
    {
        if (len == 0)
            return;

        // 确保可读数据是足够的
        assert(len <= ReadAbleSize());
        // 拷贝数据并移动读位置
        std::copy((char *)ReadPosition(), (char *)ReadPosition() + len, (char *)buf);
        // 移动读位置
        MoveReadIdx(len);
    }
    std::string ReadString(uint64_t len)
    {
        assert(len <= ReadAbleSize());
        std::string str;
        str.resize(len);
        Read(&str[0], len);
        return str;
    }
    void ReadBuffer(Buffer *buf, uint64_t len)
    {
        assert(len <= ReadAbleSize());
        Read(buf->Address(), len);
    }

    // 获取一行数据
    std::string GetLine()
    {
        char *pos = (char *)memchr(Address(), '\n', ReadAbleSize());
        if (pos == nullptr)
            return "";
        return ReadString(pos - (char *)ReadPosition() + 1);
    }

    // 清空换缓冲区
    void clear()
    {
        _buffer.clear();
        _read_idx = 0;
        _write_idx = 0;
    }
};

// 通信套接字操作封装模块----------------------------------------------------------------------------------------
static const std::string default_server_bind_ip = "0.0.0.0"; // 服务端bind的默认ip
static const int default_backlog = 10;                       // 服务器默认的listen等待处理的队列长度

class Socket
{
private:
    int _sockfd; // 套接字文件描述符

private:
    // 创建套接字
    bool CreateSocket()
    {
        _sockfd = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
        if (_sockfd < 0)
        {
            ERR_LOG("CREATE SOCKET FAIL");
            return false;
        }
        DBG_LOG("CREATE SOCKET SUCCESS");
        return true;
    }
    // 绑定ip和端口号
    bool Bind(const uint16_t port, const std::string &ip = default_server_bind_ip)
    {
        struct sockaddr_in addr;
        addr.sin_family = AF_INET;
        addr.sin_port = htons(port);
        addr.sin_addr.s_addr = inet_addr(ip.c_str()); //?
        socklen_t len = sizeof(addr);

        int ret = bind(_sockfd, (const struct sockaddr *)&addr, len);
        if (ret < 0)
        {
            ERR_LOG("BIND SOCKET FAIL");
            return false;
        }
        DBG_LOG("BIND SOCKET SUCCESS");
        return true;
    }
    // 服务器开始监听
    bool Listen(int backlog = default_backlog)
    {
        int ret = listen(_sockfd, backlog);
        if (ret < 0)
        {
            ERR_LOG("LISTEN SOCKFD FAIL");
            return false;
        }
        DBG_LOG("LISTEN SOCKFD SUCCESS.......");
        return true;
    }
    // 客户端发起连接
    bool Connect(const uint16_t port, const std::string &ip)
    {
        struct sockaddr_in addr;
        addr.sin_family = AF_INET;
        addr.sin_port = htons(port);
        addr.sin_addr.s_addr = inet_addr(ip.c_str());
        socklen_t len = sizeof(addr);

        int ret = connect(_sockfd, (const struct sockaddr *)&addr, len);
        if (ret < 0)
        {
            ERR_LOG("CLINET CONNECT FAIL");
            return false;
        }
        DBG_LOG("CLINET CONNECT SUCCESS");
        return true;
    }

public:
    // 构造和析构函数
    Socket() : _sockfd(-1) {}
    Socket(int sockfd) : _sockfd(sockfd) {}
    ~Socket() { Close(); }

    // 获取文件描述符
    int Fd() { return _sockfd; }
    // 关闭文件描述符
    void Close()
    {
        if (_sockfd != -1)
            close(_sockfd);
    }

    // 服务端处理新连接
    int Accept()
    {
        int newfd = accept(_sockfd, nullptr, nullptr);
        if (newfd < 0)
        {
            ERR_LOG("ACCEPT GET NEW CONNECTION FAIL");
            return -1;
        }
        INF_LOG("GET A NEW LINK...");
        return newfd;
    }

    // 创建一个服务端监听套接字
    bool CreateServer(const uint16_t port, const std::string &ip = default_server_bind_ip, const int backlog = default_backlog)
    {
        // 创建监听套接字
        if (CreateSocket() == false)
            return false;
        // 设置套接字为非阻塞
        SetFileNonBlock();
        // 开启端口地址重用
        SetReuseAddressAndPort();
        // 绑定ip和端口号
        if (Bind(port, ip) == false)
            return false;
        // 开始监听
        if (Listen(backlog) == false)
            return false;
        return true;
    }
    // 创建一个客户端通信套接字
    bool CreateClient(const uint16_t port, const std::string &ip)
    {
        // 创建通信套接字
        if (CreateSocket() == false)
            return false;
        // 向服务器发起连接
        if (Connect(port, ip) == false)
            return false;
        return true;
    }

    // 发送数据
    ssize_t Send(const void *data, size_t len, int flag = 0)
    {
        ssize_t ret = send(_sockfd, data, len, flag);
        if (ret < 0)
        {
            // 发送被信号打断了或者发送缓冲区空间不足
            if (errno == EAGAIN || errno == EINTR)
                return 0;
            ERR_LOG("SEND FAIL : %s", strerror(errno));
            return -1;
        }
        return ret;
    }
    // 非阻塞发送数据
    ssize_t SendNonBlock(const void *data, size_t len)
    {
        if (len == 0)
            return 0;
        return Send(data, len, MSG_DONTWAIT);
    }
    // 接收数据
    ssize_t Recv(void *data, size_t len, int flag = 0)
    {
        ssize_t ret = recv(_sockfd, data, len, flag);
        // 返回小于0的话，就是读取出问题了
        // 返回等于0的话，就是对端关闭了连接
        if (ret <= 0)
        {
            // 读取被信号打断或者接收缓冲区没有数据了
            if (errno == EINTR || errno == EAGAIN)
            {
                return 0;
            }
            // 其他问题的话，就是真正的读取出问题了
            ERR_LOG("RECV FAIL : %s", strerror(errno));
            return -1;
        }
        // 读取成功返回读取的字节数
        return ret;
    }
    // 非阻塞读取数据
    ssize_t RecvNonBlock(void *data, size_t len)
    {
        if (len == 0)
            return 0;
        return Recv(data, len, MSG_DONTWAIT);
    }

    // 设置文件属性为非阻塞
    void SetFileNonBlock()
    {
        int flag = fcntl(_sockfd, F_GETFL, 0);
        fcntl(_sockfd, F_SETFL, flag | O_NONBLOCK);
    }
    // 设置服务器地址端口重用
    void SetReuseAddressAndPort()
    {
        int val = 1;
        setsockopt(_sockfd, SOL_SOCKET, SO_REUSEADDR | SO_REUSEPORT, &val, sizeof(val));
    }
};

class Epoll;
class EventLoop;
// 对文件描述符事件监控和处理方法封装的模块----------------------------------------------------------------------------
class Channel
{
private:
    int _fd;           // 文件描述符
    uint32_t _events;  // 要监控的事件
    uint32_t _revents; // 就绪的事件
    EventLoop *_loop;  // loop模块

    using ReadCallback = std::function<void()>;
    using WriteCallback = std::function<void()>;
    using ErrorCallback = std::function<void()>;
    using CloseCallback = std::function<void()>;
    using EventCallback = std::function<void()>;
    // 各种事件触发的回调函数
    ReadCallback _read_callback;      // 读事件回调
    WriteCallback _write_callback;     // 写事件回调
    ErrorCallback _error_callback;     // 错误事件回调
    CloseCallback _close_callback;     // 关闭事件回调
    EventCallback _any_event_callback; // 任意事件的回调

public:
    // 构造和析构函数
    Channel(EventLoop *loop, int fd) : _fd(fd), _events(0), _revents(0), _loop(loop) {}
    ~Channel() = default;

    // 获取文件描述符
    int Fd() { return _fd; }
    // 获取就绪的事件--在向epoll中添加事件监控的时候，构造epoll_event结构体需要获取该变量
    uint32_t GetEvents() { return _events; }
    // 设置实际就绪的事件
    void SetRevent(uint32_t revents) { _revents = revents; }

    // 回调函数的设置
    void SetReadCallback(const ReadCallback &cb) { _read_callback = cb; }
    void SetWriteCallback(const WriteCallback &cb) { _write_callback = cb; }
    void SetErrorCallback(const ErrorCallback &cb) { _error_callback = cb; }
    void SetCloseCallback(const CloseCallback &cb) { _close_callback = cb; }
    void SetEventCallback(const EventCallback &cb) { _any_event_callback = cb; }

    // 是否监控读事件
    bool IsRead() { return _events & EPOLLIN; }
    // 是否监控写事件
    bool IsWrite() { return _events & EPOLLOUT; }
    // 开启读事件
    void EnableRead()
    {
        _events |= EPOLLIN;
        Update();
    }
    // 开启写事件
    void EnableWirte()
    {
        _events |= EPOLLOUT;
        Update();
    }
    // 关闭读事件
    void DisableRead()
    {
        _events &= EPOLLIN;
        Update();
    }
    // 关闭写事件
    void DisableWirte()
    {
        _events &= EPOLLOUT;
        Update();
    }

    // 更新事件监控
    void Update();
    // 删除事件监控
    void Remove();

    // 事件处理
    void HandleEvent()
    {
        // 可读事件就绪
        if (_revents & EPOLLIN)
        {
            if (_read_callback)
                _read_callback();
        }
        // 可写事件就绪
        if (_revents & EPOLLOUT)
        {
            if (_write_callback)
                _write_callback();
        }
        // 错误事件就绪
        else if (_revents & EPOLLERR)
        {
            if (_error_callback)
                _error_callback();
        }
        // 关闭事件就绪
        else if (_revents & EPOLLHUP)
        {
            if (_close_callback)
                _close_callback();
        }

        // 执行任意事件回调
        if (_any_event_callback)
            _any_event_callback();
    }
};

// 对Epoll封装的模块------------------------------------------------------------------------------------------------------------------
static const int MAX_EVENTS = 1024;

class Epoll
{
private:
    int _epfd; // epoll文件描述符
    struct epoll_event _evs[MAX_EVENTS];
    std::unordered_map<int, Channel *> _channels;

private:
    // 判断该文件描述符是否存在
    bool HasChannel(int fd)
    {
        auto it = _channels.find(fd);
        if (it == _channels.end())
        {
            return false;
        }
        return true;
    }
    // 对epoll监控的文件描述符进行操作
    void Update(Channel *channel, int op)
    {
        // 初始化struct epoll_event结构体
        int fd = channel->Fd();
        struct epoll_event ev;
        ev.data.fd = fd;
        ev.events = channel->GetEvents();

        // 对epoll进行操作
        int ret = epoll_ctl(_epfd, op, fd, &ev);
        if (ret < 0)
        {
            ERR_LOG("EPOLL CTL FAIL : %s", strerror(errno));
        }
    }

public:
    Epoll()
    {
        _epfd = epoll_create(MAX_EVENTS);
        if (_epfd < 0)
        {
            ERR_LOG("EPOLL CREATE FAIL");
            abort();
        }
        DBG_LOG("EPOLL CREATE SUCCESS");
    }
    ~Epoll() = default;

    // 添加、更新事件监控
    void UpdateEvent(Channel *channel)
    {
        bool ret = HasChannel(channel->Fd());
        // 不存在则添加
        if (ret == false)
        {
            // 插入到hashMap中
            _channels.insert(std::make_pair(channel->Fd(), channel));
            Update(channel, EPOLL_CTL_ADD);
        }
        // 存在则更新事件监控
        else
        {
            Update(channel, EPOLL_CTL_MOD);
        }
    }
    // 移除事件监控
    void RemoveEvent(Channel *channel)
    {
        auto it = _channels.find(channel->Fd());
        if (it == _channels.end())
            return;
        _channels.erase(it);
        Update(channel, EPOLL_CTL_DEL);
    }

    // 启动epoll服务
    void StartEpoll(std::vector<Channel *> *active)
    {
        // 阻塞式获取就绪事件
        int ret = epoll_wait(_epfd, _evs, MAX_EVENTS, -1);
        if (ret < 0)
        {
            // 被信号打断了
            if (errno == EINTR)
                return;
            ERR_LOG("EPOLL WAIT FAIL");
            abort();
        }

        // 将就绪的Channel添加到返回数组active中
        for (int i = 0; i < ret; i++)
        {
            auto it = _channels.find(_evs[i].data.fd);
            assert(it != _channels.end());
            // 更新channel的就绪事件变量
            it->second->SetRevent(_evs[i].events);
            active->push_back(it->second);
        }
    }
};

// 定时任务类--------------------------------------------------------------------------------------------------------------------------------------
using TaskFunctor = std::function<void()>;    // 定时器任务函数
using ReleaseFunctor = std::function<void()>; // 移除定时任务的函数
class TimerTask
{
private:
    uint64_t _task_id;          // 定时任务id
    int _timeout;               // 超时时间
    bool _canceled;             // 是否取消定时任务
    TaskFunctor _task_cb;       // 定时任务
    ReleaseFunctor _release_cb; // 析构的时候，用于释放该定时任务在时间轮中保存的记录

public:
    // 构造和析构函数
    TimerTask(uint64_t task_id, int delay) : _task_id(task_id), _timeout(delay), _canceled(false) {}
    ~TimerTask()
    {
        _release_cb();
        if (_canceled == false)
            _task_cb();
    }

    // 获取超时时间
    int GetTimeout() { return _timeout; }
    // 取消定时任务
    void CancelTimerTask() { _canceled = true; }
    // 设置定时任务
    void SetTaskFunctor(const TaskFunctor &cb) { _task_cb = cb; }
    // 设置移除定时任务的回调
    void SetReleaseFunctor(const ReleaseFunctor &cb) { _release_cb = cb; }
};

// 时间轮类与Timerfd结合--------------------------------------------------------------------------------------------------------------------------------------
static const int default_capacity = 60;          // 默认时间轮的容量大小
using PrtTimerTask = std::shared_ptr<TimerTask>; // 定时任务的share_ptr指针
using WeakTimerTask = std::weak_ptr<TimerTask>;  // 定时任务的weak_ptr指针

class TimerWheel
{
private:
    int _tick;                                     // 滴答指针
    int _capacity;                                 // 时间轮容量
    std::vector<std::vector<PrtTimerTask>> _wheel; // 时间轮
    std::unordered_map<int, WeakTimerTask> _tasks; // 任务id和任务的映射
    EventLoop *_loop;                              // 绑定线程

    int _timerfd;                              // 定时器文件描述符
    std::unique_ptr<Channel> _timerfd_channel; // 管理定时器文件描述符的Channel模块

private:
    void Release(uint64_t task_id)
    {
        auto it = _tasks.find(task_id);
        return;
        _tasks.erase(it);
    }

    // 创建timerfd文件描述符,并启动定时器
    static int CreateTimerfd()
    {
        // 创建定时器
        int timerfd = timerfd_create(CLOCK_MONOTONIC, 0);
        if (timerfd < 0)
        {
            ERR_LOG("TIMERFD CREATE FAIL");
            abort();
        }

        // 设置定时器的时间
        struct itimerspec itime;
        itime.it_value.tv_sec = 1;
        itime.it_value.tv_nsec = 0;
        itime.it_interval.tv_sec = 0;
        itime.it_interval.tv_nsec = 0;

        // 启动定时器
        int ret = timerfd_settime(timerfd, 0, &itime, nullptr);
        return timerfd;
    }
    // timerfd读事件就绪的回调函数
    void ReadTimerfd()
    {
        uint64_t val = 0;
        ssize_t ret = read(_timerfd, &val, sizeof(val));
        if (ret <= 0)
        {
            if (errno == EAGAIN || errno == EINTR)
                return;
            ERR_LOG("READ TIMERFD FAIL");
            abort();
        }

        for (int i = 0; i < val; i++)
        {
            RunTimerWheel();
        }
    }

    // 运行时间轮--滴答指针向后走一步
    void RunTimerWheel()
    {
        // 向前走一步
        _tick = (_tick + 1) % _capacity;
        // 释放执行超时的任务
        _wheel[_tick].clear();
    }

    // 添加定时任务
    void TimerAddInLoop(uint64_t task_id, int delay, const TaskFunctor &task_cb)
    {
        // 确定插入的位置
        int pos = (_tick + delay) % _capacity;
        // 构建PtrTimerTask
        PrtTimerTask newTask = PrtTimerTask(new TimerTask(task_id, delay));
        newTask->SetTaskFunctor(task_cb);
        newTask->SetReleaseFunctor(std::bind(&TimerWheel::Release, this, task_id));

        // 插入到哈希表中
        _tasks[task_id] = WeakTimerTask(newTask);
        // 插入到时间轮中
        _wheel[pos].push_back(newTask);
    }
    // 刷新定时任务
    void TimerRefreshInLoop(uint64_t task_id)
    {
        // 获取该任务的超时时间
        auto it = _tasks.find(task_id);
        if (it == _tasks.end())
            return;

        // 获取一个PtrTimerTask变量
        PrtTimerTask pt = it->second.lock();
        // 确定新位置
        int pos = (_tick + pt->GetTimeout()) % _capacity;
        // 将PtrTimerTask插入到新位置
        _wheel[pos].push_back(pt);
    }
    // 删除定时任务
    void TimerCancelInLoop(int task_id)
    {
        auto it = _tasks.find(task_id);
        if (it == _tasks.end())
            return;

        // 取消定时任务
        it->second.lock()->CancelTimerTask();
    }

public:
    TimerWheel(EventLoop *loop, int capacity = default_capacity)
        : _tick(0), _capacity(60), _wheel(_capacity), _loop(loop), _timerfd(CreateTimerfd()), _timerfd_channel(new Channel(_loop, _timerfd))
    {
        // 添加读事件回调函数
        _timerfd_channel->SetReadCallback(std::bind(&TimerWheel::ReadTimerfd, this));
        // 启动读事件监控
        _timerfd_channel->EnableRead();
    }
    ~TimerWheel() = default;

    // 添加定时任务
    void TimerAdd(uint64_t task_id, int delay, const TaskFunctor &task_cb);
    // 刷新定时任务
    void TimerRefresh(uint64_t task_id);
    // 取消定时任务
    void TimerCancel(int task_id);

    // 判断该任务是否在时间轮中
    bool HasTimerTask(uint64_t task_id)
    {
        auto it = _tasks.find(task_id);
        if (it == _tasks.end())
            return false;
        return true;
    }
};

// EventLoop与Eventfd结合模块--------------------------------------------------------------------------------------------------------------------------------------

class EventLoop
{
private:
    std::thread::id _thread_id; // 线程id
    Epoll _epoller;             // epoll模型
    int _eventfd;               // 进行服务器唤醒的文件描述符
    std::unique_ptr<Channel> _eventfd_channel;

    TimerWheel _timer_wheel; // 时间轮
    using TaskFunctor = std::function<void()>;
    std::vector<TaskFunctor> _tasks; // 任务队列
    std::mutex _mutex;               // 管理任务队列的锁

private:
    // 创建eventfd文件描述符
    static int CreateEventfd()
    {
        int efd = eventfd(0, EFD_CLOEXEC | EFD_NONBLOCK);
        if (efd < 0)
        {
            ERR_LOG("CREATE EVENTFD FAIL");
            abort();
        }
        return efd;
    }
    // 向eventfd中写入数据
    void WriteEventfd()
    {
        uint64_t val = 1;
        ssize_t ret = write(_eventfd, &val, sizeof(val));
        if (ret < 0)
        {
            if (errno == EINTR)
                return;
            ERR_LOG("WRITE EVENTFD FAIL");
            abort();
        }
    }
    // 读取eventfd中的数据
    void ReadEventfd()
    {
        uint64_t val = 0;
        ssize_t ret = read(_eventfd, &val, sizeof(val));
        if (ret <= 0)
        {
            if (errno == EAGAIN || errno == EINTR)
                return;
            ERR_LOG("READ EVENTFD FAIL");
            abort();
        }
    }
    // 运行任务池中的任务
    void RunTaskQueue()
    {
        std::vector<TaskFunctor> tmp_tasks;
        // 加锁
        {
            std::unique_lock<std::mutex> lock(_mutex);
            // 将任务交换出来
            tmp_tasks.swap(_tasks);
        }
        for (auto &task : tmp_tasks)
        {
            task();
        }
    }

public:
    EventLoop()
        : _thread_id(std::this_thread::get_id()), _eventfd(CreateEventfd()), _eventfd_channel(new Channel(this, _eventfd)), _timer_wheel(this)
    {
        _eventfd_channel->SetReadCallback(std::bind(&EventLoop::ReadEventfd, this));
        _eventfd_channel->EnableRead();
    }
    ~EventLoop() = default;

    // 启动epoll
    void StartEpoll()
    {
        while (1)
        {
            // 启动epoll事件监控，获取就绪事件
            std::vector<Channel *> actives;
            _epoller.StartEpoll(&actives);
            // 对就绪的事件进行处理
            for (auto &channel : actives)
            {
                channel->HandleEvent();
            }
            // 处理任务队列的事件
            RunTaskQueue();
        }
    }
    // 判断当前线程是否是loop绑定的线程
    bool IsInLoop()
    {
        return _thread_id == std::this_thread::get_id();
    }
    // 运行任务
    void RunTask(const TaskFunctor &task_cb)
    {
        if (IsInLoop())
            task_cb();
        TaskPushInQueue(task_cb);
    }
    // 将任务放入任务队列中
    void TaskPushInQueue(const TaskFunctor &task_cb)
    {
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _tasks.push_back(task_cb);
        }
        // 唤醒执行任务池中的任务
        WriteEventfd();
    }

    // 添加定时任务
    void TimerAdd(uint64_t task_id, int delay, const TaskFunctor &task_cb)
    {
        _timer_wheel.TimerAdd(task_id, delay, task_cb);
    }
    // 刷新定时任务
    void TimerRefesh(uint64_t task_id)
    {
        _timer_wheel.TimerRefresh(task_id);
    }
    // 取消定时任务
    void TimerCancel(uint64_t task_id)
    {
        _timer_wheel.TimerCancel(task_id);
    }
    // 判断是否含有该任务
    bool HasTimerTask(uint64_t task_id)
    {
        return _timer_wheel.HasTimerTask(task_id);
    }

    // 更新事件监控
    void UpdateEvent(Channel *channel)
    {
        _epoller.UpdateEvent(channel);
    }
    // 移除事件监控
    void RemoveEvent(Channel *channel)
    {
        _epoller.RemoveEvent(channel);
    }
};

// 更新事件监控
void Channel::Update()
{
    _loop->UpdateEvent(this);
}
// 删除事件监控
void Channel::Remove()
{
    _loop->RemoveEvent(this);
}

// 添加定时任务
void TimerWheel::TimerAdd(uint64_t task_id, int delay, const TaskFunctor &task_cb)
{
    _loop->RunTask(std::bind(&TimerWheel::TimerAddInLoop, this, task_id, delay, task_cb));
}
// 刷新定时任务
void TimerWheel::TimerRefresh(uint64_t task_id)
{
    _loop->RunTask(std::bind(&TimerWheel::TimerRefreshInLoop, this, task_id));
}
// 取消定时任务
void TimerWheel::TimerCancel(int task_id)
{
    _loop->RunTask(std::bind(&TimerWheel::TimerCancelInLoop, this, task_id));
}

// 创建线程模块-------------------------------------------------------------------------------------------------------------------------------------
class LoopThread
{
private:
    std::thread _thread; // 线程
    EventLoop *_loop;    // 绑定的loop模块

    std::mutex _mutex;             // 锁
    std::condition_variable _cond; // 信号量

private:
    // 线程函数
    void ThreadEntry()
    {
        // 创建loop
        EventLoop loop;
        {
            // 加锁
            std::unique_lock<std::mutex> lock(_mutex);
            _loop = &loop;
            _cond.notify_all();
        }
        // 启动loop线程的服务器
        loop.StartEpoll();
    }

public:
    // 构造和析构函数
    LoopThread()
        : _thread(std::thread(&LoopThread::ThreadEntry, this)), _loop(nullptr) {}

    // 获取loop对象
    EventLoop *GetLoop()
    {
        EventLoop *loop;
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _cond.wait(lock, [&]()
                       { return _loop != nullptr; });
            loop = _loop;
        }
        return loop;
    }
};

// 管理主从Reactor线程的线程池模块-------------------------------------------------------------------------------------------------------------------------------------
class LoopThreadPool
{
private:
    EventLoop *_baseloop; // 主线程
    int _thread_count;    // 从属线程数量
    int _next_idx;        // 下标--保证每个线程都能够均匀工作

    std::vector<LoopThread *> _threads; // 管理
    std::vector<EventLoop *> _loops;    // 保存从属线程地址的数组

public:
    LoopThreadPool(EventLoop *baseloop) : _baseloop(baseloop), _thread_count(0), _next_idx(0) {}
    // 设置线程数量
    void SetThreadCount(int num) { _thread_count = num; }
    // 创建从属线程
    void Create()
    {
        if (_thread_count > 0)
        {
            _threads.resize(_thread_count);
            _loops.resize(_thread_count);
            for (int i = 0; i < _thread_count; i++)
            {
                _threads[i] = new LoopThread();
                _loops[i] = _threads[i]->GetLoop();
            }
        }
    }
    // 获取一个线程
    EventLoop *GetLoop()
    {
        if (_thread_count == 0)
        {
            return _baseloop;
        }

        _next_idx = (_next_idx + 1) % _thread_count;
        return _loops[_next_idx];
    }
};

// 通用类型模块----------------------------------------------------------------------------------------------------------------------------------------
class Any
{
private:
    // 父类
    class placeholder
    {
    public:
        virtual ~placeholder() = 0;
        virtual const std::type_info &type() = 0;
        virtual placeholder *Clone() = 0;
    };
    // 子类
    template <class T>
    class holder : public placeholder
    {
    public:
        holder(const T &val) : _val(val) {}
        // 函数重写
        virtual ~holder() {}
        virtual const std::type_info &type() { return typeid(T); }
        virtual placeholder *Clone() { return new holder(_val); }

    public:
        T _val;
    };

public:
    // 构造函数
    Any() : _content(nullptr) {}
    template <class T>
    Any(const T &val) : _content(new holder<T>(val)) {}
    Any(const Any &other) : _content(other._content ? _content = other._content->Clone() : nullptr) {}

    // 赋值重载
    template <class T>
    Any &operator=(const T &val)
    {
        any(val).Swap(*this);
        return *this;
    }
    Any &operator=(Any other)
    {
        Swap(other);
        return *this;
    }

    // 交换函数
    void Swap(Any &other) { std::swap(_content, other._content); }
    // 获取存储的内容
    template <class T>
    T *Get()
    {
        assert(typeid(T) == _content->type());
        return &((holder<T> *)_content)->_val;
    }

private:
    placeholder *_content;
};

// 连接管理的模块------------------------------------------------------------------------------------------------------------------------------------
enum ConnStatus
{
    DISCONNECTED, // 连接关闭状态
    CONNECTING,   // 连接待初始化状态
    CONNECTED,    // 连接完毕状态
    DISCONNECTING // 连接待关闭状态
};

class Connection;
using PtrConnection = std::shared_ptr<Connection>;

class Connection : public std::enable_shared_from_this<Connection>
{
private:
    EventLoop *_loop;        // 绑定线程
    uint64_t _conn_id;       // 连接Id
    int _sockfd;             // 通信描述符
    Socket _socket;          // 通信套接字
    Channel _sockfd_channel; // 管理通信描述符

    bool _enable_inactive_release; // 是否启用超时连接销毁
    ConnStatus _status;            // 连接状态
    Buffer _in_buffer;             // 接收缓冲区
    Buffer _out_buffer;            // 发送缓冲区
    Any _content;                  // 上下文数据

    using ConnectedCallback = std::function<void(const PtrConnection &)>;
    using MessageCallback = std::function<void(const PtrConnection &, Buffer *)>;
    using ClosedCallback = std::function<void(const PtrConnection &)>;
    using AnyEventCallback = std::function<void(const PtrConnection &)>;
    // 用户的回调函数
    ConnectedCallback _connected_callback;
    MessageCallback _message_callback;
    ClosedCallback _closed_callback;
    AnyEventCallback _anyevent_callback;
    // 连接断开时系统的销毁回调函数
    ClosedCallback _server_closed_callback;

private:
    // 读事件回调函数
    void HandleRead()
    {
        char data[65535]; // 8KB
        ssize_t ret = _socket.RecvNonBlock(data, 65535);
        // 读取失败了，就关闭连接
        if (ret < 0)
        {
            ShutdownInLoop();
        }
        // 读取成功,将读取的内容放到接收缓冲区中，移动写偏移
        _in_buffer.Write(data, ret);
        // 执行用户处理数据回调函数
        if (_message_callback)
            _message_callback(shared_from_this(), &_in_buffer);
    }
    // 写事件回调
    void HandleWrite()
    {
        ssize_t ret = _socket.SendNonBlock(_out_buffer.WritePosition(), _out_buffer.ReadAbleSize());
        if (ret < 0)
        {
            // 如果有数据待处理
            if (_in_buffer.ReadAbleSize() > 0)
            {
                if (_message_callback)
                    _message_callback(shared_from_this(), &_in_buffer);
            }
            // 关闭连接
            return ReleaseInLoop();
        }
        _out_buffer.MoveReadIdx(ret);

        if (_out_buffer.ReadAbleSize() == 0)
        {
            // 关闭写事件监控
            if (_sockfd_channel.IsWrite() == true)
                _sockfd_channel.DisableWirte();

            // 关闭连接
            if (_status == DISCONNECTING)
                ReleaseInLoop();
        }
    }
    // 错误事件回调
    void HandleError()
    {
        HandleClose();
    }
    // 连接关闭回调
    void HandleClose()
    {
        // 如果有数据待处理
        if (_in_buffer.ReadAbleSize() > 0)
        {
            if (_message_callback)
                _message_callback(shared_from_this(), &_in_buffer);
        }
        ReleaseInLoop();
    }
    // 任意事件的回调函数
    void HandleEvent()
    {
        // 刷新定时任务
        if (_loop->HasTimerTask(_conn_id))
        {
            _loop->TimerRefesh(_conn_id);
        }
        // 执行用户的任意事件回调函数
        if (_anyevent_callback)
            _anyevent_callback(shared_from_this());
    }

    // 处理待发送、待处理的数据
    void ShutdownInLoop()
    {
        // 修改连接状态
        _status = DISCONNECTING;
        // 数据待处理
        if (_in_buffer.ReadAbleSize() > 0)
        {
            if (_message_callback)
                _message_callback(shared_from_this(), &_in_buffer);
        }
        // 数据待发送
        if (_out_buffer.ReadAbleSize() > 0)
        {
            // 启动写事件监控
            if (_sockfd_channel.IsWrite() == false)
                _sockfd_channel.EnableWirte();
        }
        if (_out_buffer.ReadAbleSize() == 0)
        {
            ReleaseInLoop();
        }
    }

    // 初始化连接操作
    void EstablishedInLoop()
    {
        std::cout << _status << std::endl;
        assert(_status == CONNECTING);
        // 修改连接状态
        _status = CONNECTED;

        // 启动可读事件监控
        _sockfd_channel.EnableRead();
        // 调用用户的连接成功回调函数
        if (_connected_callback)
            _connected_callback(shared_from_this());
    }

    // 真正的关闭连接操作
    void ReleaseInLoop()
    {
        // 修改连接状态
        _status = DISCONNECTED;
        // 移除事件监控
        _sockfd_channel.Remove();
        // 关闭文件描述符
        _socket.Close();
        // 取消定时任务
        if (_loop->HasTimerTask(_conn_id))
            _loop->TimerCancel(_conn_id);
        // 执行用户的关闭连接回调
        if (_closed_callback)
            _closed_callback(shared_from_this());
        // 执行系统的关闭连接回调函数
        if (_server_closed_callback)
            _server_closed_callback(shared_from_this());
    }
    // 发送数据--将数据放入发送缓冲区中，启动写事件监控
    void SendInLoop(Buffer &buf)
    {
        // 在连接准备完毕的时候才可以进行操作
        if (_status == CONNECTED)
        {
            _out_buffer.WriteBuffer(buf);
            if (_sockfd_channel.IsWrite() == false)
                _sockfd_channel.EnableWirte();
        }
    }

    // 启动非活跃连接定时销毁
    void EnableInactiveReleaseInLoop(int delay)
    {
        _enable_inactive_release = true;
        // 如果已经存在定时销毁了，就刷新一下
        if (_loop->HasTimerTask(_conn_id))
            return _loop->TimerRefesh(_conn_id);
        // 不存在则添加
        _loop->TimerAdd(_conn_id, delay, std::bind(&Connection::ReleaseInLoop, this));
    }
    // 取消非活跃连接定时销毁
    void CancelInactiveReleaseInLoop()
    {
        _enable_inactive_release = false;
        if (_loop->HasTimerTask(_conn_id))
            _loop->TimerCancel(_conn_id);
    }
    // 切换协议
    void UpgradeInLoop(const Any &content, const ConnectedCallback &conn_cb, const MessageCallback &mess_cb,
                       const ClosedCallback &close_cb, const AnyEventCallback &event_cb)
    {
        _content = content;
        _connected_callback = conn_cb;
        _message_callback = mess_cb;
        _closed_callback = close_cb;
        _anyevent_callback = event_cb;
    }

public:
    Connection(EventLoop *loop, int conn_id, int sockfd)
        : _loop(loop), _conn_id(conn_id), _sockfd(sockfd), _socket(sockfd), _sockfd_channel(loop, sockfd), _enable_inactive_release(false), _status(CONNECTING)
    {
        // 绑定回调函数
        _sockfd_channel.SetReadCallback(std::bind(&Connection::HandleRead, this));
        _sockfd_channel.SetWriteCallback(std::bind(&Connection::HandleWrite, this));
        _sockfd_channel.SetErrorCallback(std::bind(&Connection::HandleError, this));
        _sockfd_channel.SetCloseCallback(std::bind(&Connection::HandleClose, this));
        _sockfd_channel.SetEventCallback(std::bind(&Connection::HandleEvent, this));
    }

    // 设置各种回调函数
    void SetConnectedCallback(const ConnectedCallback &cb) { _connected_callback = cb; }
    void SetMessageCallback(const MessageCallback &cb) { _message_callback = cb; }
    void SetClosedCallback(const ClosedCallback &cb) { _closed_callback = cb; }
    void SetAnyEventCallback(const AnyEventCallback &cb) { _anyevent_callback = cb; }
    void SetServerClosedCallback(const ClosedCallback &cb) { _server_closed_callback = cb; }

    // 获取文件描述符
    int Fd() { return _sockfd; }
    // 获取连接Id
    uint64_t Id() { return _conn_id; }
    // 设置上下文数据
    void SetContent(const Any &content) { _content = content; }
    // 获取上下文数据
    Any *GetConntent() { return &_content; }
    // 是否处于CONNECTED状态
    bool IsConnected() { return _status == CONNECTED; }

    // 初始化连接
    void Established()
    {
        _loop->RunTask(std::bind(&Connection::EstablishedInLoop, this));
    }
    // 提供给用户的关闭连接的接口，并非实际关闭，而是将各项内容处理完之后，在执行关闭
    void Shutdown()
    {
        _loop->RunTask(std::bind(&Connection::ShutdownInLoop, this));
    }
    // 发送数据接口
    void Send(const char *data, size_t len)
    {
        Buffer buf;
        buf.Write(data, len);
        _loop->RunTask(std::bind(&Connection::SendInLoop, this, buf));
    }
    // 启动非活跃连接定时销毁
    void EnableInactiveRelease(int delay)
    {
        _loop->RunTask(std::bind(&Connection::EnableInactiveReleaseInLoop, this, delay));
    }
    // 取消非活跃连接定时销毁
    void CancelInactiveRelease()
    {
        _loop->RunTask(std::bind(&Connection::CancelInactiveReleaseInLoop, this));
    }
    // 切换协议
    void Upgrade(const Any &content, const ConnectedCallback &conn_cb, const MessageCallback &mess_cb,
                 const ClosedCallback &close_cb, const AnyEventCallback &event_cb)
    {
        assert(_loop->IsInLoop());
        _loop->RunTask(std::bind(&Connection::UpgradeInLoop, this, content, conn_cb, mess_cb, close_cb, event_cb));
    }
};

// 监听套接字模块-----------------------------------------------------------------------------------------------------------------------------
class Acceptor
{
private:
    EventLoop *_baseloop;    // 主线程模块
    Socket _socket;          // 监听套接字模块
    Channel _listen_channel; // 管理监听套接字描述符模块

    // 新连接到来的回调函数
    using AcceptCallback = std::function<void(int)>;
    AcceptCallback _accept_callback;

private:
    // 创建服务端套接字
    int CreateServer(uint16_t port)
    {
        bool ret = _socket.CreateServer(port);
        if (ret == false)
        {
            abort();
        }
        return _socket.Fd();
    }
    // 服务端读事件就绪回调
    void HandleRead()
    {
        // 获取文件描述符
        int newfd = _socket.Accept();
        if (newfd < 0)
            return;
        // 执行回调函数
        if (_accept_callback)
            _accept_callback(newfd);
    }

public:
    Acceptor(EventLoop *baseloop, uint16_t port) : _baseloop(baseloop), _socket(CreateServer(port)), _listen_channel(_baseloop, _socket.Fd())
    {
        _listen_channel.SetReadCallback(std::bind(&Acceptor::HandleRead, this));
    }
    // 设置回调函数
    void SetAcceptCallback(const AcceptCallback &cb) { _accept_callback = cb; }
    // 启动服务，获取连接
    void ListenLoop() { _listen_channel.EnableRead(); }
};

// 服务器汇总模块----------------------------------------------------------------------------------------------------------------------------
class TcpServer
{
private:
    int _conn_id_up; // 自增的conn_id
    int _timeout;    // 超时时间
    uint64_t _port;  // 端口号

    EventLoop _baseloop;           // 主线程
    Acceptor _acceptor;            // 监听套接字
    LoopThreadPool _pool;          // 从属线程池
    bool _enable_inactive_release; // 是否启用非活跃连接销毁

    std::unordered_map<int, PtrConnection> _conns; // 管理连接

    using ConnectedCallback = std::function<void(const PtrConnection &)>;
    using MessageCallback = std::function<void(const PtrConnection &, Buffer *)>;
    using ClosedCallback = std::function<void(const PtrConnection &)>;
    using AnyEventCallback = std::function<void(const PtrConnection &)>;
    using Functor = std::function<void()>;
    // 用户的回调函数
    ConnectedCallback _connected_callback;
    MessageCallback _message_callback;
    ClosedCallback _closed_callback;
    AnyEventCallback _anyevent_callback;

private:
    // 为新连接创建一个Connection对象
    void NewConnection(int sockfd)
    {
        _conn_id_up++;
        // 创建一个Connection对象
        PtrConnection conn(new Connection(_pool.GetLoop(), _conn_id_up, sockfd));
        // 设置回调函数
        conn->SetConnectedCallback(_connected_callback);
        conn->SetMessageCallback(_message_callback);
        conn->SetClosedCallback(_closed_callback);
        conn->SetAnyEventCallback(_anyevent_callback);
        conn->SetServerClosedCallback(std::bind(&TcpServer::Remove, this, std::placeholders::_1));
        // 启动非活跃连接销毁
        if (_enable_inactive_release)
            conn->EnableInactiveRelease(_timeout);
        // 初始化连接
        conn->Established();

        _conns.insert(std::make_pair(_conn_id_up, conn));
    }

    // 移除连接
    void RemoveInLoop(const PtrConnection &conn)
    {
        auto it = _conns.find(conn->Id());
        if (it != _conns.end())
        {
            _conns.erase(conn->Id());
        }
    }
    void Remove(const PtrConnection &conn)
    {
        _baseloop.RunTask(std::bind(&TcpServer::RemoveInLoop, this, conn));
    }
    // 添加定时任务
    void TimerTaskAddInLoop(const Functor &task_cb, int delay)
    {
        _conn_id_up++;
        _baseloop.TimerAdd(_conn_id_up, delay, task_cb);
    }

public:
    // 构造函数
    TcpServer(int port)
        : _conn_id_up(0), _timeout(0), _port(port), _acceptor(&_baseloop, port), _pool(&_baseloop), _enable_inactive_release(false)
    {
        _acceptor.SetAcceptCallback(std::bind(&TcpServer::NewConnection, this, std::placeholders::_1));
        _acceptor.ListenLoop();
    }

    // 设置各种回调函数
    void SetConnectedCallback(const ConnectedCallback &cb) { _connected_callback = cb; }
    void SetMessageCallback(const MessageCallback &cb) { _message_callback = cb; }
    void SetClosedCallback(const ClosedCallback &cb) { _closed_callback = cb; }
    void SetAnyEventCallback(const AnyEventCallback &cb) { _anyevent_callback = cb; }

    // 设置线程数量
    void SetLoopThreadCount(int num) { _pool.SetThreadCount(num); }
    // 启动非活跃连接销毁
    void EnableInactiveRelease(int timeout)
    {
        _timeout = timeout;
        _enable_inactive_release = true;
    }
    // 启动非活跃连接销毁
    void CancelInactiveRelease(int timeout)
    {
        _timeout = 0;
        _enable_inactive_release = false;
    }
    // 添加定时任务
    void TimerTaskAdd(const Functor &task_cb, int delay)
    {
        _baseloop.RunTask(std::bind(&TcpServer::TimerTaskAddInLoop, this, task_cb, delay));
    }

    // 启动服务器
    void Start()
    {
        // 创建线程
        _pool.Create();
        // 启动获取就绪事件
        _baseloop.StartEpoll();
    }
};